package com.alibaba.datax.plugin.reader.mongodbreader.util;

import com.alibaba.datax.common.element.*;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.reader.mongodbreader.KeyConstant;
import com.alibaba.datax.plugin.reader.mongodbreader.MongoDBReaderErrorCode;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.google.common.base.Joiner;
import com.google.common.base.Strings;
import com.mongodb.client.MongoCursor;
import org.apache.commons.lang3.StringUtils;
import org.bson.Document;
import org.bson.types.ObjectId;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.Iterator;

/**
 * @author cuijianpeng
 * @date 2020/6/25 10:33
 */
public class MongoUtil {

    /**
     * 解析QueryFilter
     *
     * @return Document
     */
    public static Document parseQueryFilter(Configuration config) {
        final boolean isObjectId = config.getBool(KeyConstant.IS_OBJECTID, true);
        final String query = config.getString(KeyConstant.MONGO_QUERY);
        final String queryFeature = config.getString(KeyConstant.MONGO_QUERY_FEATURE);
        final Object lowerBound = config.get(KeyConstant.LOWER_BOUND);
        final Object upperBound = config.get(KeyConstant.UPPER_BOUND);
        if (lowerBound == null || upperBound == null) {
            throw DataXException.asDataXException(MongoDBReaderErrorCode.ILLEGAL_VALUE,
                    MongoDBReaderErrorCode.ILLEGAL_VALUE.getDescription());
        }
        Document filter = new Document();
        if (lowerBound.equals(KeyConstant.MIN)) {
            if (!upperBound.equals(KeyConstant.MAX)) {
                filter.append(KeyConstant.MONGO_PRIMARY_ID, new Document(KeyConstant.LT, isObjectId ? new ObjectId(upperBound.toString()) : upperBound));
            }
        } else {
            if (upperBound.equals(KeyConstant.MAX)) {
                filter.append(KeyConstant.MONGO_PRIMARY_ID, new Document(KeyConstant.GTE, isObjectId ? new ObjectId(lowerBound.toString()) : lowerBound));
            } else {
                filter.append(KeyConstant.MONGO_PRIMARY_ID, new Document(KeyConstant.GTE, isObjectId ? new ObjectId(lowerBound.toString()) : lowerBound).append(KeyConstant.LT, isObjectId ? new ObjectId(upperBound.toString()) : upperBound));
            }
        }
        if (!Strings.isNullOrEmpty(query)) {
            Feature feature = null;
            if(StringUtils.isNotBlank(queryFeature)){
                feature = FastJsonUtil.getJsonFeature(queryFeature);
            }
            final JSONObject queryObject = FastJsonUtil.parseObject(query, feature);
            Document queryFilter = new Document(queryObject);
            filter = new Document(KeyConstant.AND, Arrays.asList(filter, queryFilter));
        }
        return filter;
    }

    public static void sendToWrite(Configuration config, RecordSender recordSender, MongoCursor<Document> dbCursor) {
        final JSONArray columnMeta = JSON.parseArray(config.getString(KeyConstant.MONGO_COLUMN));
        if (columnMeta == null) {
            throw DataXException.asDataXException(MongoDBReaderErrorCode.ILLEGAL_VALUE,
                    MongoDBReaderErrorCode.ILLEGAL_VALUE.getDescription());
        }
        while (dbCursor.hasNext()) {
            Document item = dbCursor.next();
            Record record = createRecord(recordSender, columnMeta, item);
            recordSender.sendToWriter(record);
        }
    }

    private static Record createRecord(RecordSender recordSender, JSONArray mongodbColumnMeta, Document item) {
        Record record = recordSender.createRecord();
        Iterator columnItera = mongodbColumnMeta.iterator();
        while (columnItera.hasNext()) {
            JSONObject column = (JSONObject) columnItera.next();
            Object tempCol = item.get(column.getString(KeyConstant.COLUMN_NAME));
            if (tempCol == null) {
                if (KeyConstant.isDocumentType(column.getString(KeyConstant.COLUMN_TYPE))) {
                    String[] name = column.getString(KeyConstant.COLUMN_NAME).split("\\.");
                    if (name.length > 1) {
                        Object obj;
                        Document nestedDocument = item;
                        for (String str : name) {
                            obj = nestedDocument.get(str);
                            if (obj instanceof Document) {
                                nestedDocument = (Document) obj;
                            }
                        }

                        if (null != nestedDocument) {
                            Document doc = nestedDocument;
                            tempCol = doc.get(name[name.length - 1]);
                        }
                    }
                }
            }
            if (tempCol == null) {
                //continue; 这个不能直接continue会导致record到目的端错位
                record.addColumn(new StringColumn(null));
            } else if (tempCol instanceof Double) {
                //TODO deal with Double.isNaN()
                record.addColumn(new DoubleColumn((Double) tempCol));
            } else if (tempCol instanceof Boolean) {
                record.addColumn(new BoolColumn((Boolean) tempCol));
            } else if (tempCol instanceof Date) {
                record.addColumn(new DateColumn((Date) tempCol));
            } else if (tempCol instanceof Integer) {
                record.addColumn(new LongColumn((Integer) tempCol));
            } else if (tempCol instanceof Long) {
                record.addColumn(new LongColumn((Long) tempCol));
            } else {
                if (KeyConstant.isArrayType(column.getString(KeyConstant.COLUMN_TYPE))) {
                    String splitter = column.getString(KeyConstant.COLUMN_SPLITTER);
                    if (Strings.isNullOrEmpty(splitter)) {
                        throw DataXException.asDataXException(MongoDBReaderErrorCode.ILLEGAL_VALUE,
                                MongoDBReaderErrorCode.ILLEGAL_VALUE.getDescription());
                    } else {
                        ArrayList array = (ArrayList) tempCol;
                        String tempArrayStr = Joiner.on(splitter).join(array);
                        record.addColumn(new StringColumn(tempArrayStr));
                    }
                } else {
                    record.addColumn(new StringColumn(tempCol.toString()));
                }
            }
        }
        return record;
    }
}
